package com.atguigu.pro1


import java.sql.Timestamp
import java.util.UUID

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.source.{RichSourceFunction, SourceFunction}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

import scala.util.Random


/**
 * @description: app下载渠道分析
 * @time: 2021/4/1 14:41
 * @author: baojinlong
 * */
object AppMarketByChannel {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 方便测试全局并行度为1
    env.setParallelism(1)
    // 设置时间语义为事件时间
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val dataStream: DataStream[MarketUserBehavior] = env.addSource(new SimulatedSource)
      .assignAscendingTimestamps(_.timestamp)

    // 统计1h或者10minutes,每隔1s
    val resultStream: DataStream[MarketViewCount] = dataStream
      .filter(item => !"uninstall".equals(item.behavior))
      // 根据渠道和行为分组
      .keyBy(data => (data.channel, data.behavior))
      // 统计的是过去一天时间的数据,每隔5s统计一次
      .timeWindow(Time.days(1), Time.seconds(5))
      .process(new MarketCountByChannel)
    resultStream.print("resultStream")
    env.execute("jobTest")
  }
}

// 定义输入数据样例类
case class MarketUserBehavior(userId: String, behavior: String, channel: String, timestamp: Long)

// 自定义测试数据源
class SimulatedSource extends RichSourceFunction[MarketUserBehavior] {
  // 是否运行标志位
  var running: Boolean = true
  // 定义用户行为和渠道的集合
  val behaviorSet: Seq[String] = Seq("view", "download", "install", "uninstall")
  val behaviorSetSize: Int = behaviorSet.size
  val channelSet: Seq[String] = Seq("appstore", "weibo", "wechat", "tieba")
  val channelSetSize: Int = channelSet.size
  // 生成数据的最大值
  val maxCountValue: Long = Long.MaxValue
  // 计数变量
  var countIndex = 0L

  override def run(sourceContext: SourceFunction.SourceContext[MarketUserBehavior]): Unit = {
    while (running && countIndex < maxCountValue) {
      val userId: String = UUID.randomUUID.toString
      val behavior: String = behaviorSet(Random.nextInt(behaviorSetSize))
      val channel: String = channelSet(Random.nextInt(channelSetSize))
      val timestamp: Long = System.currentTimeMillis()
      sourceContext.collect(MarketUserBehavior(userId, behavior, channel, timestamp))
      countIndex += 1
      Thread.sleep(50)
    }
  }

  override def cancel(): Unit = running = false
}

// 定义输出数据样例类报表展示
case class MarketViewCount(windowStart: String, windowEnd: String, channel: String, behavior: String, count: Long)

// 自定义窗口函数,这是使用的是全窗口函数
class MarketCountByChannel extends ProcessWindowFunction[MarketUserBehavior, MarketViewCount, (String, String), TimeWindow] {
  // 当前窗口触发计算的时候调用的方法
  override def process(key: (String, String), context: Context, elements: Iterable[MarketUserBehavior], out: Collector[MarketViewCount]): Unit = {
    val windowStart: String = new Timestamp(context.window.getStart).toString
    val windowEnd: String = new Timestamp(context.window.getEnd).toString
    val channel: String = key._1
    val behavior: String = key._2
    val count: Long = elements.size
    out.collect(MarketViewCount(windowStart, windowEnd, channel, behavior, count))
  }
}


